#include "tpool/threadspool.h"
#include "headers.h"
#include "our_header.h"
#include <stdio.h>
#include <stdlib.h>

static TPoll one_tpoll = {0};
static bool isInitPoll = false;


int initThreadsPoll_h(int pthreads_sum, int queen_caipacity) {
	if(pthreads_sum > MAX_THREADS_NUM || pthreads_sum < 1)
		return -1;
	if(queen_caipacity > MAX_QUEEN_CPTY || queen_caipacity < 1)
		return -1;

	one_tpoll.pthreads_sum = pthreads_sum;
	one_tpoll.pthreads_rm_num = pthreads_sum;
	one_tpoll.share_queen_data.array_queen_caipacity = queen_caipacity;

	pthread_mutex_init(&one_tpoll.share_queen_data.mutex, NULL);
	pthread_cond_init(&one_tpoll.share_queen_data.cond, NULL);
	one_tpoll.share_queen_data.flag = 0;
	one_tpoll.pthread_tid =
		(pthread_t *)calloc(pthreads_sum, sizeof(pthread_t));
	// 加入mysql 的连接
	one_tpoll.thread_gets = calloc(pthreads_sum, sizeof(tdata_t));
	for(int i = 0; i < pthreads_sum; i++) {
		if(getConnect_7th(&one_tpoll.thread_gets[i].connect)) {
			fprintf(stderr, "init mysql connct error\n");
			exit(1);
		}
		one_tpoll.thread_gets[i].shareQueen=&one_tpoll.share_queen_data;
	}

	isInitPoll = init_queue_h(queen_caipacity);
	if(isInitPoll == 0)
		isInitPoll = 1;
	return isInitPoll ? 0 : -1;
}

int produceSocketFd_h(const SQItem *train) {
	int statu = -1;
	if(isInitPoll) {
		pthread_mutex_lock(&one_tpoll.share_queen_data.mutex);
		if((statu = enqueue_h(train)) != 0) {
			printf("queen full\n");
			pthread_cond_broadcast(&one_tpoll.share_queen_data.cond);
		}
		pthread_cond_signal(&one_tpoll.share_queen_data.cond);
		pthread_mutex_unlock(&one_tpoll.share_queen_data.mutex);
	}
	return statu;
}

int consumSocketFd_h(SQItem *trian) {
	int tag_sta = 0;
	if(isInitPoll) {
		pthread_mutex_lock(&one_tpoll.share_queen_data.mutex);
		while(dequeue_h(trian) && one_tpoll.share_queen_data.flag == 0) {
			pthread_cond_wait(&one_tpoll.share_queen_data.cond, &one_tpoll.share_queen_data.mutex);
		}
		if(one_tpoll.share_queen_data.flag)
			tag_sta = 1;
		pthread_mutex_unlock(&one_tpoll.share_queen_data.mutex);
	}
	return tag_sta;
}

int exitThreadsPoll_h() {
	if(isInitPoll) {
		isInitPoll = 0;
		pthread_mutex_lock(&one_tpoll.share_queen_data.mutex);
		one_tpoll.share_queen_data.flag = 1;
		pthread_cond_broadcast(&one_tpoll.share_queen_data.cond);
		pthread_mutex_unlock(&one_tpoll.share_queen_data.mutex);
		destoryqueun_h();
		printf("eixtThreasPoll %d,%d\n", one_tpoll.pthreads_sum, one_tpoll.pthreads_rm_num);

		for(int i = 0; i < one_tpoll.pthreads_sum - one_tpoll.pthreads_rm_num;
			i++) {
			int tag = pthread_join(one_tpoll.pthread_tid[i], NULL);
			if(tag)
				fprintf(stdout, "pthread_join,%s", strerror(tag));
		}
		free(one_tpoll.pthread_tid);
		for(int i = 0; i < one_tpoll.pthreads_sum; i++) {
			closeConnect_7th(one_tpoll.thread_gets[i].connect);
		}
		printf("wait over\n");
	}
	return 0;
}

int startThreadsPoll_h(int threads_num, void *(*func)(void *)) {
	if(!isInitPoll)
		return -1;
	int count = 0;
	int last_pthread_index = one_tpoll.pthreads_sum - one_tpoll.pthreads_rm_num;
	for(int i = last_pthread_index;
		i < threads_num + last_pthread_index && i < one_tpoll.pthreads_sum;
		i++) {

		if(pthread_create(&one_tpoll.pthread_tid[i], NULL, func,&one_tpoll.thread_gets[i])) {
			perror("startThreadsPoll_h pthread_create");
			break;
		}
		count++;
	}
	one_tpoll.pthreads_rm_num -= count;
	return count;
}
